package com.xiaomaoguai;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

/**
 * @author WeiHui-Z
 * @version v1.0.0
 * @date 2019/8/30 20:28
 * @since JDK 1.8
 */
@Slf4j
public class BlcsMain {

	private static final String JOB_NAME = "keplerTagProccess";

	//kafka.server=10.139.52.45:9092,10.139.52.57:9092,10.139.52.49:9092
   //	kafka.groupId=es-test-16
	//kafka.keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer
//	kafka.valueDeserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer

	public static void main(String[] args) throws Exception {
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		//checkpointing时间间隔为1分钟
		env.enableCheckpointing(60 * 1000);
		env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
		env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
		// 重启次数 + 重启时间间隔
		env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, org.apache.flink.api.common.time.Time.seconds(10L)));
		//window的时间采用事件时间
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

		Properties props = new Properties();
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.139.52.45:9092,10.139.52.57:9092,10.139.52.49:9092");
		props.put(ConsumerConfig.GROUP_ID_CONFIG, "es-test-16");
		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

		//设置输入source
		FlinkKafkaConsumer011 kafkaConsumerSource = new FlinkKafkaConsumer011(
				"fcp-kepler-201905161412",
				new DecisionApplyDtoDeserializationSchema(),
				props);

		//类型转换
		DataStreamSource dataStreamSource = env.addSource(kafkaConsumerSource).setParallelism(1);
		//dataStreamSource.addSink(new BlcsSinkToMySQL()); //数据 sink 到 mysql
		env.execute(JOB_NAME);
		log.info("flink job={} start", JOB_NAME);
	}

}
